package day2;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.util.UUID;

public class Test5_source {
    public static void main(String[] args) throws Exception {
        // 需求：每隔一秒生成一条订单信息

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // 读数据叫 addSource 写数据叫 addSink
        env.addSource(new SourceFunction<Order>() {
            @Override
            public void run(SourceContext<Order> out) throws Exception {
                Random random = new Random();
                while (true) {
                    Order order = new Order();
                    order.setUserId(random.nextInt(3));
                    order.setMoney(random.nextInt(101));
                    order.setOrderId(UUID.randomUUID().toString());
                    order.setTs(System.currentTimeMillis());
                    out.collect(order);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        }).print();

        env.execute();
    }
}
